/*
* Copyright (c) Huawei Technologies Co., Ltd. 2020-2021. All rights reserved.
* gazelle is licensed under the Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*     http://license.coscl.org.cn/MulanPSL2
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY OR FIT FOR A PARTICULAR
* PURPOSE.
* See the Mulan PSL v2 for more details.
*/

#include <securec.h>

#include <rte_ring.h>
#include <rte_mbuf.h>
#include "common/dpdk_common.h"
#include "lstack_log.h"
#include "lstack_dispatch_ring.h"

#define DISPATCH_RING_SIZE  256
#define DISPATCH_RING_COUNT 64

struct dispatch_ring {
    struct rte_ring *ring;
    uint64_t rx_drop;
};

static struct dispatch_ring g_dispatch_ring[DISPATCH_RING_COUNT];

int dispatch_ring_create(uint16_t queue_id, unsigned socket_id)
{
    unsigned flags = RING_F_SC_DEQ;
    char ring_name[RTE_MEMZONE_NAMESIZE] = {0};
    snprintf_s(ring_name, sizeof(ring_name), sizeof(ring_name) - 1, "dispatch_ring_%d", queue_id);

    g_dispatch_ring[queue_id].ring = rte_ring_create(ring_name, DISPATCH_RING_SIZE, socket_id, flags);
    if (g_dispatch_ring[queue_id].ring == NULL) {
        LSTACK_LOG(ERR, LSTACK, "dispatch_ring_create %s  failed\n", ring_name);
        return -1;
    }

    return 0;
}

static struct dispatch_ring *dispatch_ring_lookup(uint16_t queue_id)
{
    if (g_dispatch_ring[queue_id].ring == NULL) {
        char ring_name[RTE_MEMZONE_NAMESIZE] = {0};
        snprintf_s(ring_name, sizeof(ring_name), sizeof(ring_name) - 1, "dispatch_ring_%d", queue_id);
        g_dispatch_ring[queue_id].ring = rte_ring_lookup(ring_name);
    }

    return &g_dispatch_ring[queue_id];
}

int dispatch_ring_enqueue(uint16_t queue_id, struct rte_mbuf *m)
{

    struct dispatch_ring *ring = dispatch_ring_lookup(queue_id);
    if (ring->ring == NULL) {
        rte_pktmbuf_free(m);
        ring->rx_drop++;
        return 0;
    }
    if (rte_ring_enqueue(ring->ring, m) < 0) {
        rte_pktmbuf_free(m);
        ring->rx_drop++;
    } else {
        mbuf_private_set_queue(m, queue_id);
    }
    return 0;
}

unsigned int dispatch_ring_dequeue_burst(uint16_t queue_id, struct rte_mbuf **pkts)
{
    struct dispatch_ring *ring = dispatch_ring_lookup(queue_id);
    if (ring->ring == NULL) {
        return 0;
    }

    return rte_ring_dequeue_burst(ring->ring, (void **)pkts, 128, NULL);
}
